package com.spzx.common.rabbit.config;

import com.alibaba.fastjson2.JSON;
import com.spzx.common.rabbit.entity.GuiguCorrelationData;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.EventListener;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.lang.Nullable;
import org.springframework.stereotype.Component;

import java.util.concurrent.TimeUnit;

/**
 * 服务器启动时，执行rabbitTemplate初始化，设置确认函数和回退函数
 *   ApplicationEvent      一些子事件的父类。
 *   ApplicationReadyEvent 具体子事件。表示应用程序启动好，IOC容器初始化好，存在相关bean对象了。再进行相关的初始化。
 *   也可以使用相关的注解替代： @EventListener
 */
@Slf4j
@Component
public class RabbitInitConfigApplicationListener implements ApplicationListener<ApplicationReadyEvent> {

    @Autowired
    RabbitTemplate rabbitTemplate;

    @Autowired
    RedisTemplate redisTemplate;


    @Override
    public void onApplicationEvent(ApplicationReadyEvent event) {
        System.out.println("*****************  event = " + event);
        this.setUpInitRabbitTemplete();
    }


    private void setUpInitRabbitTemplete() {
        //1.设置确认回调函数
        //交换机是否接收到消息都会执行确认回调。
        rabbitTemplate.setConfirmCallback((correlationData, ack, cause)->{
            //关联数据
            log.info("correlationData="+correlationData);
            if(ack){ //交换机收到消息了
                log.info("ack="+ack);
            }else{ //交换机没收到消息
                log.info("ack="+ack);
                log.error("cause="+cause);

                //重发消息
                this.retrySendMsg(correlationData);
            }
        });

        //2.设置回退回调函数
        //交换机收到了，但是无法路由到队列。那么会执行回退函数。
        rabbitTemplate.setReturnsCallback(returned->{
            //returned=ReturnedMessage [message=(Body:'hello' MessageProperties [
            // headers={spring_returned_message_correlation=mq:a30942cfaef541488963a69f2888cf77},
            // contentType=text/plain, contentEncoding=UTF-8, contentLength=0, receivedDeliveryMode=PERSISTENT,
            // priority=0, deliveryTag=0]), replyCode=312, replyText=NO_ROUTE, exchange=spzx.test, routingKey=xxxx]
            log.error("returned="+returned);
            log.error("Returned: " + returned.getMessage() + "\nreplyCode: " + returned.getReplyCode()
                    + "\nreplyText: " + returned.getReplyText() + "\nexchange/rk: "
                    + returned.getExchange() + "/" + returned.getRoutingKey());

            //returned对象头信息中会含有关联数据的id.关联数据并没有退回来。
            String correlationId = returned.getMessage().getMessageProperties().getHeader("spring_returned_message_correlation");
            String guiguCorrelationDataStr = (String)redisTemplate.opsForValue().get(correlationId);
            GuiguCorrelationData guiguCorrelationData = JSON.parseObject(guiguCorrelationDataStr,GuiguCorrelationData.class);

            //延迟插件，会将消息退回，执行重试。这里不进行重试（方案1）。
            boolean delay = guiguCorrelationData.isDelay();
            if(delay){ //判断是否为延迟消息，延迟消息则不进行重试
                return;
            }

            this.retrySendMsg(guiguCorrelationData);
        });

    }


    /**
     * 消息重新发送
     *
     * @param correlationData
     */
    private void retrySendMsg(CorrelationData correlationData) {
        GuiguCorrelationData guiguCorrelationData = (GuiguCorrelationData)correlationData;
        int retryCount = guiguCorrelationData.getRetryCount();
        if(retryCount >= 3){
            log.error("重试消息达到最大次数3次。停止重试。"); //可以给管理员发消息，可以将消息存储到数据库...
            return;
        }

        retryCount++;
        guiguCorrelationData.setRetryCount(retryCount);
        //更新缓存数据
        redisTemplate.opsForValue().set(guiguCorrelationData.getId(), JSON.toJSONString(correlationData),10, TimeUnit.MINUTES);

        boolean delay = guiguCorrelationData.isDelay();
        log.info("retrySendMsg - delay="+delay);
        if(!delay){ //不是延迟消息，重发消息

            rabbitTemplate.convertAndSend(guiguCorrelationData.getExchange(),
                    guiguCorrelationData.getRoutingKey(),
                    guiguCorrelationData.getMessage(),
                    guiguCorrelationData);
        }else{ //是延迟消息，重发消息(方案2)。
            rabbitTemplate.convertAndSend(guiguCorrelationData.getExchange(),
                    guiguCorrelationData.getRoutingKey(),
                    guiguCorrelationData.getMessage(),
                    new MessagePostProcessor(){
                        @Override
                        public Message postProcessMessage(Message message) throws AmqpException {
                            message.getMessageProperties().setDelay(guiguCorrelationData.getDelayTime()*1000); //单位毫秒
                            return message;
                        }
                    },correlationData);
        }

        log.info("进行消息重发！");
    }
}
